<?php
require_once '/data/wwwroot/package/vendor/autoload.php';

Class CoreKafka
{

    protected string $broker_list = '';
    protected int $refresh_interval_ms = 10000;

    protected function readBrokerList(): string
    {
        $this->broker_list = '172.25.17.84:9092';
    }

    protected function producerConfig()
    {
        $config = \Kafka\ProducerConfig::getInstance();
        $config->setMetadataRefreshIntervalMs($this->refresh_interval_ms);
        $config->setMetadataBrokerList($this->broker_list);
        $config->setBrokerVersion('1.0.0');
        $config->setRequiredAck(1);
        $config->setIsAsyn(false);
        $config->setProduceInterval(500);
    }

    protected function consumerConfig()
    {
        $config = \Kafka\ConsumerConfig::getInstance();
        $config->setMetadataRefreshIntervalMs($this->refresh_interval_ms);
        $config->setMetadataBrokerList($this->broker_list);
        $config->setGroupId('test');
        $config->setBrokerVersion('1.0.0');
        $config->setTopics(['test']);
    }
}


$producer = new \Kafka\Producer(
    function() {
        return [
            [
                'topic' => 'test',
                'value' => 'test....message.',
                'key' => 'testkey',
            ],
        ];
    }
);
//$producer->setLogger($logger);
$producer->success(function($result) {
    var_dump($result);
});
$producer->error(function($errorCode) {
    var_dump($errorCode);
});
$producer->send(true);




//$config->setOffsetReset('earliest');
$consumer = new \Kafka\Consumer();
//$consumer->setLogger($logger);
$consumer->start(function($topic, $part, $message) {
    var_dump($message);
});